package org.zhadoop.streaming.util.datatype;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.TimeZone;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.streaming.config.StreamingConfig;
import org.zhadoop.streaming.exception.StreamingException;

import com.google.common.base.Strings;

public class TimestampParser implements DataTypeParser {

	/**
	 * 
	 */
	private static final long serialVersionUID = -9051894477118042630L;
	private static final Logger LOG = LoggerFactory.getLogger(TimestampParser.class);
	
	/**
	 * 默认时区
	 * 该参数一般由客户端在提交任务之前设置一次
	 */
	private SimpleDateFormat outputFormatter;
	
	/**
	 * timestamp 类型允许输入带毫秒，纳秒的数字
	 * 所以数字精确到秒级的
	 */
	private SimpleDateFormat inputFormatter;
	
	/**
	 * 带timezone的输入格式
	 */
	private SimpleDateFormat inputFormatterTZ;
	
	/**
	 * 带ms的输入格式
	 */
	private SimpleDateFormat inputFormatterMS;
	
	/**
	 * 带timezone和 ms 的输入格式
	 */
	private SimpleDateFormat inputFormatterMSTZ;
	
	private final TimeZone timeZone;
	
	public TimestampParser(StreamingConfig config) throws StreamingException {
		timeZone = TimeZoneUtils.getTimeZone(config);
		initDefaultInputFormat(timeZone);
        initOutputFormat(timeZone);
	}
	
	private void initOutputFormat(TimeZone timeZone) {
		outputFormatter = new SimpleDateFormat(TimeConstants.TIMESTAMP_MSTIME_TIMEZONE_FORMAT);
		outputFormatter.setLenient(false);
		outputFormatter.setTimeZone(timeZone);
	}

	private void initDefaultInputFormat(TimeZone timeZone) {
		inputFormatter = new SimpleDateFormat(TimeConstants.TIMESTAMP_FORMAT);
		inputFormatter.setLenient(false);
		inputFormatter.setTimeZone(timeZone);
		
		inputFormatterTZ = new SimpleDateFormat(TimeConstants.TIMESTAMP_TIMEZONE_FORMAT);
		inputFormatterTZ.setLenient(false);
		inputFormatterTZ.setTimeZone(timeZone);
		
		inputFormatterMS = new SimpleDateFormat(TimeConstants.TIMESTAMP_MSTIME_FORMAT);
        inputFormatterMS.setLenient(false);
        inputFormatterMS.setTimeZone(timeZone);

        inputFormatterMSTZ = new SimpleDateFormat(TimeConstants.TIMESTAMP_MSTIME_TIMEZONE_FORMAT);
        inputFormatterMSTZ.setLenient(false);
        inputFormatterMSTZ.setTimeZone(timeZone);
	}

	@Override
	public Object createValue(String value) throws StreamingException {
		if(Strings.isNullOrEmpty(value)){
			return null;
		}
		try {
			Date date = parseDateFormatFromInput(value).parse(value);
			return new Timestamp(date.getTime());
		} catch (Exception e) {
			LOG.warn("Failed to create {} instance.", Timestamp.class.getName());
			throw new StreamingException("Failed to create " + Timestamp.class.getName() + " instance.");
		}
	}
	
	@Override
	public String toStringValue(Object value) throws StreamingException {
		if(null == value){
			return null;
		}
		return outputFormatter.format(value);
	}

	private SimpleDateFormat parseDateFormatFromInput(String input) {
		//点好出现的索引，如果没有出现点号，则说明不包含毫秒
		int pointIndex = -1;
		
		//空格出现的次数，如果有两个空格，则说明带了时区
		int blankCount = 0;
		
		char[] inputChars = input.toCharArray();
		for(int i=0; i < inputChars.length; i++){
			switch (inputChars[i]) {
			case '.':
				pointIndex = i;
				break;
			case ' ':
				blankCount ++;
				break;
			default:
				break;
			}
		}
		
		//带毫秒
		if(pointIndex != -1){
			//带时区
			if(blankCount == 2){
				return inputFormatterMSTZ;
			}else{
				return inputFormatterMS;
			}
		}else{
			//带时区
            if (blankCount == 2)
            {
                return inputFormatterTZ;
            }
            else
            {
                return inputFormatter;
            }
		}
	}



	public TimeZone getTimeZone() {
		return timeZone;
	}


}
